package com.inspinia.efks.config;

import com.github.danielwegener.logback.kafka.KafkaAppender;
import com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy;
import com.github.danielwegener.logback.kafka.encoding.LayoutKafkaMessageEncoder;
import com.github.danielwegener.logback.kafka.keying.HostNameKeyingStrategy;
import net.logstash.logback.fieldnames.ShortenedFieldNames;
import net.logstash.logback.layout.LogstashLayout;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.PropertySource;

import java.nio.charset.Charset;

/**
 * @author Veblen
 */
@PropertySource(value = "kafka-elk.properties", ignoreResourceNotFound = true)
public class KafkaElkAppender extends KafkaAppender{

    private static final String CUSTOMFIELDS = "{system:test}";

    private static final String SERVERS = "bootstrap.servers";

    @Value("${topic}")
    private String TOPIC;

    @Value("${producer}")
    private String PRODUCER;

    public KafkaAppender kafkaAppender(){
        KafkaAppender kafkaAppender = new KafkaAppender();
        kafkaAppender.setTopic(TOPIC);
        kafkaAppender.setEncoder(messageEncoder());
        kafkaAppender.setKeyingStrategy(new HostNameKeyingStrategy());
        kafkaAppender.setDeliveryStrategy(new AsynchronousDeliveryStrategy());
        kafkaAppender.addProducerConfigValue(SERVERS,PRODUCER);
        return kafkaAppender;
    }

    /**
     * 消息编码
     * @return
     */
    public LayoutKafkaMessageEncoder messageEncoder(){
        LayoutKafkaMessageEncoder layoutKafkaMessageEncoder = new LayoutKafkaMessageEncoder();
        layoutKafkaMessageEncoder.setLayout(logstashLayout());
        layoutKafkaMessageEncoder.setCharset(Charset.defaultCharset());
        return layoutKafkaMessageEncoder;
    }

    /**
     * logstash
     * @return
     */
    public LogstashLayout logstashLayout(){
        LogstashLayout logstashLayout = new LogstashLayout();
        logstashLayout.setCustomFields(CUSTOMFIELDS);
        logstashLayout.setIncludeContext(true);
        logstashLayout.setIncludeCallerData(true);
        logstashLayout.setFieldNames(new ShortenedFieldNames());
        return logstashLayout;
    }
}
